package mq

import (
	"charging_pile/devices/internal/svc"
	"charging_pile/util/protocol/jiuxing"
	"charging_pile/util/protocol/mq"
	"encoding/json"

	"github.com/nats-io/nats.go"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/service"
	"github.com/zeromicro/go-zero/core/threading"
)

type natsClient struct {
	conn   *nats.Conn
	SvcCtx *svc.ServiceContext
	AckCh  map[string]chan jiuxing.Package
	//	RedisClient *redis.Redis
}

type TopicHandle func(*natsClient, []byte)

type MsgHandler func(*natsClient, *jiuxing.Package) *jiuxing.Package

var HandlerMap = make(map[uint8]MsgHandler)

func MustNewNatsConn(svcCtx *svc.ServiceContext) service.Service {
	//输入nats服务的地址和端口
	localnc := &natsClient{
		// conn:   natsConn,
		SvcCtx: svcCtx,
		AckCh:  make(map[string]chan jiuxing.Package),
	}
	addr := svcCtx.Config.MqAddr
	natsConn, err := nats.Connect(addr, nats.ReconnectHandler(func(nc *nats.Conn) {
		localnc.conn = nc
		localnc.MqSubscribeTopic(localnc.SvcCtx.LocalTopic, GroupTopicHandler)
		localnc.MqSubscribeGroupTopic("server", "device", GroupTopicHandler)
	}))
	if err != nil {
		panic(err)
	}
	localnc.conn = natsConn

	svcCtx.Mq = localnc
	return localnc
}

// 最多只有一个接收者能成功接收
func (nc *natsClient) MqSubscribeGroupTopic(topic, group string, cb TopicHandle) {
	nc.conn.QueueSubscribe(topic, group, func(msgData *nats.Msg) {
		cb(nc, msgData.Data)
	})
}

// 都能收到订阅的消息
func (nc *natsClient) MqSubscribeTopic(topic string, cb TopicHandle) {
	nc.conn.Subscribe(topic, func(msgData *nats.Msg) {
		cb(nc, msgData.Data)
	})
}

func (nc *natsClient) Publish(topic string, payload []byte) {
	nc.conn.Publish(topic, payload)
}

func (nc *natsClient) Start() {
	// topic := nc.SvcCtx.LocalTopic
	nc.MqSubscribeTopic(nc.SvcCtx.LocalTopic, GroupTopicHandler)
	nc.MqSubscribeGroupTopic("server", "device", GroupTopicHandler)
	BatchHand(nc.SvcCtx)
}

func (nc *natsClient) Stop() {
	nc.conn.Close()
}

func GroupTopicHandler(nc *natsClient, b []byte) {
	threading.RunSafe(func() {
		msg := &mq.Msg{}
		err := json.Unmarshal(b, msg)
		if err != nil {
			return
		}

		req := jiuxing.Package{}

		err = json.Unmarshal(msg.Content, &req)
		if err != nil {
			return
		}
		logx.Infof("jiuxing msg %+v\n", req)
		if handle, ok := HandlerMap[req.Cmd]; ok {
			jxMsg := handle(nc, &req)

			if jxMsg == nil {
				return
			}

			m, _ := json.Marshal(jxMsg)

			mqMsg := mq.Msg{}

			mqMsg.From = nc.SvcCtx.LocalTopic
			mqMsg.Sn = msg.Sn     //string(jxMsg.Sn)
			mqMsg.Cmd = mq.Normal //mq.MqCmd(jxMsg.Cmd)
			mqMsg.Content = m

			b, _ := json.Marshal(mqMsg)

			nc.Publish(msg.From, b)
			logx.Infof("publish:%+v", mqMsg)
		}
	})
}

func (nc *natsClient) SendMsgWaitAck(msg *jiuxing.Package) chan jiuxing.Package {
	mqMsg := mq.Msg{}

	mqMsg.From = nc.SvcCtx.LocalTopic

	tt, _ := json.Marshal(mqMsg)

	// nc.SvcCtx

	nc.Publish("", tt)
	ch := make(chan jiuxing.Package)

	nc.AckCh[""] = ch

	return ch
}
